package queue

import (
	"errors"
	"github.com/upmio/horus-agent/model"
	"github.com/upmio/horus-polymerizer/log"
	"sync"
)

//MetricQueue is exported
type MetricQueue struct {
	curlen     int
	maxlen     int
	datas      chan []*model.MetricPoint
	lock       *sync.Mutex
	name       string
	discardNum int
}

var (
	once                sync.Once
	globalmetricQue     *MetricQueue
	globalKafkaMericQue *MetricQueue
)

//NewMetricQueue is exported
func NewMetricQueue(qlen, discardNum int) (*MetricQueue, error) {
	if qlen < discardNum+100 {
		return nil, errors.New("the qlen must bigger than discardNum+200")
	}

	return &MetricQueue{
		curlen:     0,
		maxlen:     qlen,
		datas:      make(chan []*model.MetricPoint, qlen),
		lock:       new(sync.Mutex),
		name:       "defaultMetric",
		discardNum: discardNum,
	}, nil
}

//GetDefaultKafkaMetricQue is exported
func GetDefaultKafkaMetricQue() *MetricQueue {

	once.Do(func() {
		if globalKafkaMericQue == nil {
			globalKafkaMericQue = &MetricQueue{
				curlen:     0,
				maxlen:     20000,
				datas:      make(chan []*model.MetricPoint, 20000),
				lock:       new(sync.Mutex),
				name:       "defaultKafkaMetric",
				discardNum: 500,
			}
		}
	})

	return globalKafkaMericQue

}

//InitMetricQueue is exported
func InitMetricQueue(qlen, discardNum int) error {
	var err error
	globalmetricQue, err = NewMetricQueue(qlen, discardNum)
	if err != nil {
		return err
	}

	//init  globalKafkaMericQue
	que := GetDefaultKafkaMetricQue()
	if que == nil {
		return errors.New("KafkaMetricQue init fail")
	}

	return err
}

//PutMetricPoint is exported
func PutMetricPoint(metricslice []*model.MetricPoint) error {
	return globalmetricQue.PutMetric(metricslice)
}

//GetMetricSlice is exported
func GetMetricSlice() ([]*model.MetricPoint, error) {
	return globalmetricQue.GetMetric()
}

//GetMetricQcurlen is exported
func GetMetricQcurlen() int {
	return globalmetricQue.GetCurLen()
}

//GetMetricQmaxlen is exported
func GetMetricQmaxlen() int {
	return globalmetricQue.GetMaxLen()
}

//GetCurLen is exported
func (q *MetricQueue) GetCurLen() int {
	return q.curlen
}

//GetMaxLen is exported
func (q *MetricQueue) GetMaxLen() int {
	return q.maxlen
}

//GetDiscardNum is exported
func (q *MetricQueue) GetDiscardNum() int {
	return q.discardNum
}

func (q *MetricQueue) subCurlen(num int) {
	q.lock.Lock()
	q.curlen -= num
	q.lock.Unlock()
}

func (q *MetricQueue) addCurlen(num int) {
	q.lock.Lock()
	q.curlen += num
	q.lock.Unlock()
}

func (q *MetricQueue) delMetricPoint(num int) {
	for i := 0; i < num; i++ {
		_, _ = q.GetMetric()
	}
	log.Warn("metric queue  full,del some.", "now len:", q.GetCurLen())
}

//PutMetric is exported
//放数据到该队列中
func (q *MetricQueue) PutMetric(data []*model.MetricPoint) error {

	q.addCurlen(1)
	if q.curlen >= q.maxlen {
		q.delMetricPoint(q.discardNum)
	}
	q.datas <- data

	//fmt.Println("metricque curlen::::: ", q.GetCurLen(), "   evenque maxlen:::", q.GetMaxLen())
	log.Debug("metricque curlen::::: ", q.GetCurLen(), "   evenque maxlen:::", q.GetMaxLen())
	return nil
}

//GetMetric is exported
//it will block，which  I need.
func (q *MetricQueue) GetMetric() ([]*model.MetricPoint, error) {

	//if have no data ,it will block, which  I need.
	data := <-q.datas

	//should not happen
	if data == nil {
		log.Warn("get nil metirc data from queue")
		//	return nil, errors.New("get error")
	}

	//it doesnt matter that after get data..
	q.subCurlen(1)
	return data, nil
}
